package delayqueue

import (
	"github.com/go-redis/redis"
	"github.com/sirupsen/logrus"
	"strconv"
	"strings"
	"time"
)

const (
	MaxCount     = 10000
	DefaultCount = 1
	DefaultSleep = 1 * time.Second
)

type manager struct {
	queue       string
	job         delayFunc
	concurrency int
	retryCount  int32
	timeout     time.Duration
	pollSize    int64
	startedAt  int64
}

type delayFunc func(members []interface{})

// pullTasks ...
func (m *manager) pullTasks(count int64) ([]redis.Z, error) {
	opt := redis.ZRangeBy{
		Min:    "0",
		Max:    strconv.FormatInt(time.Now().Local().UnixNano(), 10),
		Offset: 0,
		Count:  count,
	}
	return Config.RedisCli.ZRangeByScoreWithScores(m.queue, opt).Result()
}

// consumerTasks ...
func (m *manager) consumerTasks(members []interface{}) (int64, error) {
	return Config.RedisCli.ZRem(m.queue, members...).Result()
}

// pushTasks ...
func (m *manager) pushTasks(members []redis.Z) error {
	return Config.RedisCli.ZAdd(m.queue, members...).Err()
}

// lenTasks ...
func (m *manager) lenTasks() (int64, error) {
	return Config.RedisCli.ZCard(m.queue).Result()
}

// allTasks ...
func (m *manager) allTasks() ([]redis.Z, error) {
	currentLen, _ := m.lenTasks()
	return Config.RedisCli.ZRangeWithScores(m.queue, 0, currentLen).Result()
}

// getDelayName ...
func (m *manager) getDelayName() string {
	return strings.Replace(m.queue, Config.Namespace+"queue:", "", 1)
}

// getPollSize ...
func (m *manager) getPollSize() {
	if m.pollSize > MaxCount {
		m.pollSize = MaxCount
	}

	if m.pollSize == 0 {
		m.pollSize = DefaultCount
	}
}

// consumer ...
func (m *manager) consumer() {
	logrus.Infof("[%d]%s delay start...", GetGoroutineID(), m.getDelayName())
	for {
		tasks, err := m.pullTasks(m.pollSize)
		if err != nil || len(tasks) == 0 {
			time.Sleep(DefaultSleep)
			continue
		}

		var members []interface{}
		for _, task := range tasks {
			members = append(members, task.Member)
		}
		if count, err := m.consumerTasks(members); err == nil && count > 0 {
			m.job(members)
		}
	}
}
